package com.xy.kafka.transaction;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;
import java.util.stream.IntStream;

public class KafkaProducerTransaction {
    public static void main(String[] args) {
        KafkaProducer<String, String> producer = buildProducer();

        //初始化事务
        producer.initTransactions();
        try {
            //开启事务
            producer.beginTransaction();
            IntStream.range(0, 10).forEach( i -> {
                if (i == 8) {
                    int j = 10/0;
                }
                ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "key" + i, "valuettttt" + i);
                producer.send(record);
                producer.flush();
            });
            //提交事务
            producer.commitTransaction();
        } catch (Exception e) {
            System.out.println("出错了:" + e.getMessage());
            //终止事务
            producer.abortTransaction();
        }
        producer.close();
    }

    private static KafkaProducer<String, String> buildProducer() {
        //1.创建kafkaAdminClient
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.72.15:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //配置事务ID，必须唯一
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-id-" + UUID.randomUUID().toString());

        //设置kafka批处理大小
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1024);
        //等待5毫秒，如果batch的数据不足 1024
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);

        //配置kafka重试机制与幂等性
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 20000);

        return new KafkaProducer<>(props);
    }
}
